/*
异步工作器的实现：
    异步工作使用双缓冲区思想，外界可以通过实现的Push接口，将任务数据添加到输入缓冲区中。
    一旦创建异步工作器，会自动创建一个线程，同时检测(条件变量等待)输入缓冲区是否有数据
    一旦检测到输入缓冲区中有数据，就会交换缓冲区，然后处理任务数据(使用注册的回调函数处理任务数据)
*/

#include <mutex>
#include <condition_variable> // 条件变量
#include <thread>
#include <functional>
#include <atomic>
#include "Buffer.hpp"

#ifndef __M_LOOPER_H__
#define __M_LOOPER_H__

namespace LogModule
{
    using Func_t = std::function<void(Buffer &)>;
    enum class AsyncType
    {
        ASYNC_SAFE,  // 阻塞
        ASYNC_UNSAFE // 扩容
    };
    class AsyncLooper
    {
    public:
        using ptr = std::shared_ptr<AsyncLooper>;
        // std::thread(&AsyncLooper::ThreadEntry, *this)的作用为：
        // 立刻创建一个函数，并且立刻执行ThreadEntry函数调用，因为该函数就是线程入口
        // 必须提供： Func_t cb 对缓冲区数据进行处理的回调函数
        AsyncLooper(const Func_t &cb, AsyncType looper_type = AsyncType::ASYNC_SAFE)
            : _stop(false),
              _thread(std::thread(&AsyncLooper::ThreadEntry, this)),
              _looper_type(looper_type),
              _callback(cb) {}
        ~AsyncLooper() { Stop(); }
        // 异步工作器停止工作
        void Stop()
        {
            // 可能有多个线程同时使用stop判断，所以将stop设为原子的
            _stop = true;
            // 异步工作器停止，要在指令结束之后停止，所以要唤醒所有的消费者
            _cond_con.notify_all();
            // 消费者处理完数据之后，就可以回收线程了
            _thread.join();
        }
        // 向生产缓冲区输入数据 -- 提供任务
        void Push(const char *data, size_t len) 
        {
            // 1. 写入缓冲区，先判断缓冲区中是否有空间，这里进行阻塞/扩容的控制
            std::unique_lock<std::mutex> lock(_mutex);
            // 写入操作，可以在条件变量处判断
            // 如果要写入的长度超过了可以写入的大小，就在条件变量出阻塞
            if (_looper_type == AsyncType::ASYNC_SAFE)
                _cond_pro.wait(lock, [&]()
                               { return (len <= _pro_buffer.WriteAbleSize()); });
            // 2. 走到这里，说明一定有空间可以写入 -- 执行写入操作
            _pro_buffer.Push(data, len);
            // 3. 生产缓冲区有数据了，就可以提醒消费者执行任务了
            _cond_con.notify_one();
        }

    private:
        // 线程入口函数 -- 线程一旦创建，就走该函数
        // 对消费缓冲区中的数据进行处理，处理完毕之后，初始化缓冲区，交换缓冲区
        void ThreadEntry()
        {
            while (!_stop)
            {
                // 加括号，如果锁出了括号，就会释放
                {
                    // 1. 处理任务，先判断生产缓冲区中有没有任务，有任务，交换缓冲区
                    //    没有任务，需要阻塞。消费缓冲区中一旦有任务，就肯定会处理完任务
                    std::unique_lock<std::mutex> lock(_mutex);
                    if(_stop == true && _pro_buffer.Empty()) break;
                    _cond_con.wait(lock, [&]()
                                   { return (_stop || !_pro_buffer.Empty()); });
                    // 2. 走到这里，说明生产缓冲区中有任务，交换缓冲区
                    _pro_buffer.Swap(_con_buffer);
                    // 3. 一旦交换完缓冲区，就可以提醒生产者，可以生产了
                    if (_looper_type == AsyncType::ASYNC_SAFE)
                        _cond_pro.notify_all();
                }
                // 4. 有了任务，处理任务 -- 回调函数
                _callback(_con_buffer);
                // 5. 处理完任务，初始化消费缓冲区
                _con_buffer.Reset();
            }
        }

    private:
        // 具体对缓冲区中数据进行处理的回调函数，由异步工作器使用者传入
        Func_t _callback;

    private:
        std::atomic<bool> _stop; // 工作器的停止标志
        Buffer _pro_buffer;      // 生产缓冲区 -- 生产者向里面生产
        Buffer _con_buffer;      // 消费缓冲区 -- 消费者从里面获取
        std::mutex _mutex;
        std::condition_variable _cond_pro; // 生产者条件变量 -- 生产者的唤醒/阻塞
        std::condition_variable _cond_con; // 消费者条件变量 -- 消费者的唤醒/阻塞
        std::thread _thread;               // 异步工作器对应的工作线程
        AsyncType _looper_type;            // 安全控制
    };
}

#endif